package store

import (
	"encoding/json"
	"fmt"
	"github.com/streadway/amqp"
	"time"
)

const GdrawUserAnswerTopic = "GdrawUserAnswerTopic"
const ConsumePowerTopic = "ConsumePowerTopic"
const AddLxChanTopic = "AddLxChanTopic"
const WxPayNotice = "WxPayNotice"
const LogerTopic = "LogerTopic"

type Rabbitmq struct {
	conf Config
	Log  Log
}

func (s *Rabbitmq) Get() *amqp.Connection {
	rabbitMqOnce.Do(func() {
		var (
			err error
		)
		rabbitMqClient, err = amqp.Dial(s.conf.Get().Rabbitmq.Addr)
		if err != nil {
			panic(err)
		}
	})
	return rabbitMqClient
}

func (s *Rabbitmq) Consume(name string, handler func(c *amqp.Delivery)) {
	s.Log.Get().Debug("mq Consume:%s", name)
	defer func() {
		if err := recover(); err != nil {
			time.Sleep(3 * time.Second)
			s.Consume(name, handler)
		}
	}()
	conn, err := amqp.Dial(s.conf.Get().Rabbitmq.Addr) //mq连接地址
	if err != nil {
		s.Log.Get().Error("mq Consume amqp.Dial err:%v", err)
		return
	}
	defer conn.Close()
	ch, err := conn.Channel()
	if err != nil {
		s.Log.Get().Error("mq Consume conn.Channel err:%v", err)
		return
	}
	defer ch.Close()

	q, err := ch.QueueDeclare(
		name,  // name
		false, // durable
		false, // delete when unused
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	if err != nil {
		s.Log.Get().Error("mq Consume ch.QueueDeclar err:%v", err)
		return
	}

	closeChan := make(chan *amqp.Error, 1)
	notifyClose := ch.NotifyClose(closeChan) //一旦消费者的channel有错误，产生一个amqp.Error，channel监听并捕捉到这个错误
	closeFlag := false
	msgs, err := ch.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false, nil)
	if err != nil {
		s.Log.Get().Error("mq Consume ch.Consume err:%v", err)
		return
	}
	for {
		select {
		case e := <-notifyClose:
			fmt.Println(e)
			close(closeChan)
			time.Sleep(5 * time.Second)
			s.Consume(name, handler)
			closeFlag = true
		case msg := <-msgs:
			handler(&msg)
		}
		if closeFlag {
			break
		}
	}
}

func (s *Rabbitmq) Publish(name string, data interface{}) (err error) {
	mq := s.Get()
	mqChan, err := mq.Channel()
	if err != nil {
		return
	}
	defer mqChan.Close()
	body, err := json.Marshal(data)
	if err != nil {
		return
	}
	err = mqChan.Publish(
		"",    // exchange
		name,  // routing key
		false, // mandatory
		false, // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        body,
			Timestamp:   time.Now(),
		})
	return
}
